package com.ctw.mapmycity.hbase.repository;

import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Repository;

import com.ctw.mapmycity.model.PostModel;

@Repository
public class PostRepository {

	@Autowired
	private HbaseTemplate hbaseTemplate;

	private String tableName = "post";

	public static String CF_INFO_STR = "cfInfo";
	public static String CF_SHARE_STR = "a";
	public static String CF_AGREE_STR = "x";

	public static byte[] CF_INFO = Bytes.toBytes(CF_INFO_STR);
	public static byte[] CF_SHARE = Bytes.toBytes(CF_SHARE_STR);
	public static byte[] CF_AGREE = Bytes.toBytes(CF_AGREE_STR);

	// column family cfInfo
	private byte[] qUserEmail = Bytes.toBytes("userEmail");
	private byte[] qDescription = Bytes.toBytes("description");
	private byte[] qTypeKey = Bytes.toBytes("typeKey");
	private byte[] qType = Bytes.toBytes("type");
	private byte[] qBoRegistered = Bytes.toBytes("boRegistered");

	private byte[] qTime = Bytes.toBytes("time");
	private byte[] qDate = Bytes.toBytes("date");

	// column family a:
	private byte[] qLatitude = Bytes.toBytes("latitude");
	private byte[] qLongitude = Bytes.toBytes("longitude");
	private byte[] qCountry = Bytes.toBytes("country");
	private byte[] qState = Bytes.toBytes("state");
	private byte[] qCity = Bytes.toBytes("city");
	private byte[] qAddress = Bytes.toBytes("address");

	// column family x:
	private byte[] qShare = Bytes.toBytes("share");
	private byte[] qConfirmation = Bytes.toBytes("confir");

	public List<PostModel> findAll() {
		Scan scan = new Scan();
		scan.addFamily(CF_INFO);
		//		scan.addFamily(CF_AGREE);

		return hbaseTemplate.find(tableName, scan, new RowMapper<PostModel>() {
			@Override
			public PostModel mapRow(Result result, int rowNum) throws Exception {

				return map(result);
			}
		});
	}

	public List<PostModel> find(String[] opts) {
		Scan scan = new Scan();
		scan.addFamily(CF_INFO);

		FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
		for (String opt : opts) {
			filterList.addFilter(new SingleColumnValueFilter(
					CF_INFO,
					qTypeKey,
					CompareOp.EQUAL,
					Bytes.toBytes(Integer.parseInt(opt))
					));
		}
		scan.setFilter(filterList);
		
		return hbaseTemplate.find(tableName, scan, new RowMapper<PostModel>() {
			@Override
			public PostModel mapRow(Result result, int rowNum) throws Exception {

				return map(result);
			}
		});
	}

	public PostModel get(String rowName) {
		return hbaseTemplate.get(tableName, rowName, new RowMapper<PostModel>() {
			@Override
			public PostModel mapRow(Result result, int rowNum) throws Exception {
				return map(result);
			}
		});

	}

	public PostModel map(Result result) {
		return new PostModel(
				Bytes.toString(result.getRow()), 
				Bytes.toString(result.getValue(CF_INFO, qUserEmail)),
				Bytes.toString(result.getValue(CF_INFO, qDescription)),
				Bytes.toInt(result.getValue(CF_INFO, qTypeKey)),
				Bytes.toString(result.getValue(CF_INFO, qType)),
				Bytes.toBoolean(result.getValue(CF_INFO, qBoRegistered)),
				Bytes.toDouble(result.getValue(CF_INFO, qLatitude)),
				Bytes.toDouble(result.getValue(CF_INFO, qLongitude)),
				Bytes.toString(result.getValue(CF_INFO, qCountry)),
				Bytes.toString(result.getValue(CF_INFO, qState)),
				Bytes.toString(result.getValue(CF_INFO, qCity)),
				Bytes.toString(result.getValue(CF_INFO, qAddress)),
				0,0, // TODO FIX it, map reduce ?
				//    			Bytes.toInt(result.getValue(CF_X, qShareCount)),
				//    			Bytes.toInt(result.getValue(CF_X, qAgreementCount)),
				Bytes.toString(result.getValue(CF_INFO, qTime)),
				Bytes.toString(result.getValue(CF_INFO, qDate))
				);
	}

	public PostModel save(final PostModel post) {
		return hbaseTemplate.execute(tableName, new TableCallback<PostModel>() {
			public PostModel doInTable(HTableInterface table) throws Throwable {
				Put p = new Put(Bytes.toBytes(post.getId()));
				p.add(CF_INFO, qUserEmail, Bytes.toBytes(post.getUserEmail()));
				p.add(CF_INFO, qDescription, Bytes.toBytes(post.getDescription()));
				p.add(CF_INFO, qTypeKey, Bytes.toBytes(post.getTypeKey()));
				p.add(CF_INFO, qType, Bytes.toBytes(post.getType()));
				p.add(CF_INFO, qBoRegistered, Bytes.toBytes(post.isBoRegistered()));
				p.add(CF_INFO, qTime, Bytes.toBytes(post.getTime()));
				p.add(CF_INFO, qDate, Bytes.toBytes(post.getDate()));

				p.add(CF_INFO, qLatitude, Bytes.toBytes(post.getLatitude()));
				p.add(CF_INFO, qLongitude, Bytes.toBytes(post.getLongitude()));
				p.add(CF_INFO, qCountry, Bytes.toBytes(post.getCountry()));
				p.add(CF_INFO, qState, Bytes.toBytes(post.getState()));
				p.add(CF_INFO, qCity, Bytes.toBytes(post.getCity()));
				p.add(CF_INFO, qAddress, Bytes.toBytes(post.getAddress()));

				//				p.add(CF_X, qShareCount, Bytes.toBytes(post.getShareCount()));
				//				p.add(CF_X, qAgreementCount, Bytes.toBytes(post.getAgreementCount()));

				table.put(p);

				return post;
			}
		});
	}

	public void delete (final String ts) {
		hbaseTemplate.execute(tableName, new TableCallback<Void>() {
			public Void doInTable(HTableInterface hti) throws Throwable {
				Delete deleteOp = new Delete(Bytes.toBytes(ts));
				hti.delete(deleteOp);
				return null;
			}
		});
	}

	public String confirm (final String ts, final String email) {
		return hbaseTemplate.execute(tableName, new TableCallback<String>() {
			public String doInTable(HTableInterface table) throws Throwable {
				Put p = new Put(Bytes.toBytes(ts));
				p.add(CF_AGREE, Bytes.toBytes(email), Bytes.toBytes(email));

				table.put(p);

				return "email";
			}
		});
	}

	public List<String> getConfirmationList(final String ts) {

		return hbaseTemplate.get(tableName, ts, new RowMapper<List<String>>() {
			@Override
			public List<String> mapRow(Result result, int rowNum) throws Exception {
				NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(CF_AGREE);
				List<String> list = new ArrayList<String>();

				for(byte[] bQunitifer : familyMap.keySet()) {
					list.add(Bytes.toString(bQunitifer));
				}

				return list;
			}
		});		
	}
}
